/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bff.gaia.examples.java.graph.util;

import com.bff.gaia.api.common.functions.*;
import com.bff.gaia.api.common.restartstrategy.RestartStrategies;
import com.bff.gaia.api.common.time.Time;
import com.bff.gaia.api.java.DataSet;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.api.java.aggregation.Aggregations;
import com.bff.gaia.api.java.functions.FunctionAnnotation;
import com.bff.gaia.api.java.operators.AggregateOperator;
import com.bff.gaia.api.java.operators.IterativeDataSet;
import com.bff.gaia.api.java.tuple.Tuple1;
import com.bff.gaia.api.java.tuple.Tuple2;
import com.bff.gaia.api.java.utils.ParameterTool;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.core.AbstractCompensationFunc;
import com.bff.gaia.core.fs.FileSystem;
import com.bff.gaia.core.fs.Path;
import com.bff.gaia.util.Collector;

import java.io.Serializable;
import java.net.URI;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import static com.bff.gaia.api.java.aggregation.Aggregations.SUM;

public class PageRankCompensationFunc extends AbstractCompensationFunc {

	private static final double DAMPENING_FACTOR = 0.85;
	private ParameterTool params;

	@Override
	public void compensate() throws Exception {

		if (this.params.has("output")) {
			final String path = this.params.get("output");
			final FileSystem fileSystem = FileSystem.get(new URI(path));
			fileSystem.delete(new Path(path), true);
		}
		final int numPages = this.params.getInt("numPages", PageRankData.getNumberOfPages());
		final int maxIterations = this.params.getInt("maxIterations", 3000);

		//task2.3_iteration
		final double EPSILON = params.getDouble("EPSILON", 1.0E-20);
//		final int killIterations = params.getInt("killIterations", -1); // 在哪一次迭代kill
//		final String killNodes = params.get("killNodes", null); // kill的节点名称
		//end

		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().setGlobalJobParameters(this.params);
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(5L, TimeUnit.SECONDS)));

		//task2.3_iteration
//		env.getConfig().setCompensationKillIterations(killIterations);
//		env.getConfig().setCompensationKillNodes(killNodes);
		env.getConfig().disableCompentionFunc();
		//end

		env.setParallelism(20);

		DataSet<Long> pagesInput = getPagesDataSet(env, this.params);
		DataSet<Tuple2<Long, Long>> linksInput = getLinksDataSet(env, params);

		final DataSet<Tuple2<Long, Long>> mapPages = pagesInput.map(new MapFunction<Long, Tuple2<Long, Long>>() {
			@Override
			public Tuple2<Long, Long> map(final Long value) throws Exception {
				return new Tuple2<Long, Long>(-1L, value);
			}
		});

		DataSet<Tuple2<Long, Double>> pagesWithRanks = (DataSet<Tuple2<Long, Double>>)pagesInput.map(new RankAssigner(1.0 / numPages));

		if (this.params.has("compensationDataPath")) {
			final DataSet<Tuple2<Long, Double>> compensationData = getCompensationdata(env, this.params);

			final AggregateOperator<Tuple2<Long, Double>> out = compensationData.aggregate(Aggregations.SUM, 1);

			double r_nonfailed = out.collect().get(0).f1; // R_nofailed
			final long n_nonfailed = compensationData.count(); // N_nonfailed

//			System.out.println(r_nonfailed + " " + n_nonfailed);

			pagesWithRanks = pagesWithRanks.leftOuterJoin(compensationData).where(0).equalTo(0).with(new PointAssigner(r_nonfailed, n_nonfailed, numPages));
		}

		final DataSet<Tuple2<Long, Long[]>> adjacencyListInput = linksInput.groupBy(0).reduceGroup(new BuildOutgoingEdgeList());

		final IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);

		DataSet<Tuple2<Long, Double>> newRanks = iteration
			// join pages with outgoing edges and distribute rank
			// 把自己的权值配分给邻居节点
			.leftOuterJoin(adjacencyListInput).where(0).equalTo(0).with(new JoinFunction<Tuple2<Long, Double>, Tuple2<Long, Long[]>, Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>>>() {
				@Override
				public Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>> join(final Tuple2<Long, Double> first, final Tuple2<Long, Long[]> second) throws Exception {
					return new Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>>(first, second);
				}
			}).flatMap(new JoinVertexWithEdgesMatch(numPages)).groupBy(0).aggregate(SUM, 1).leftOuterJoin(mapPages).where(0).equalTo(0).with(new JoinFunction<Tuple2<Long, Double>, Tuple2<Long, Long>, Tuple2<Long, Double>>() {
				@Override
				public Tuple2<Long, Double> join(final Tuple2<Long, Double> first, final Tuple2<Long, Long> second) throws Exception {
					if (second != null) {
						return new Tuple2<Long, Double>(second.f1, first.f1);
					}
					return first;
				}
			})
			// collect and sum ranks
			.groupBy(0).aggregate(SUM, 1)
			// apply dampening factor
			.map(new Dampener(DAMPENING_FACTOR, numPages));
//			.map(new PageRankRevise_1.GetIterationNums());

		final DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(newRanks, newRanks.join((DataSet<Tuple2<Long, Double>>)iteration).where(0).equalTo(0).filter(new EpsilonFilter(EPSILON)));
		final String jobName = new String("Basic Page Rank Recovery");

		if (this.params.has("output")) {
			finalPageRanks.writeAsCsv(this.params.get("output"), "\n", ",", FileSystem.WriteMode.OVERWRITE);
			env.execute(jobName);
		} else {
			System.out.println("Printing result to stdout. Use --output to specify output path.");
			finalPageRanks.print();
		}
	}

	public PageRankCompensationFunc(final String[] args) {
		super(args);
		this.params = ParameterTool.fromArgs(args);
	}

	private static DataSet<Long> getPagesDataSet(final ExecutionEnvironment env, final ParameterTool params) {
		if (!params.has("path")) {
			return env.readCsvFile(params.get("pages")).fieldDelimiter(",").lineDelimiter("\n").types(Long.class).map(new MapFunction<Tuple1<Long>, Long>() {
				@Override
				public Long map(final Tuple1<Long> v) {
					return v.f0;
				}
			});
		}
		System.out.println("Executing PageRank example with default pages data set.");
		System.out.println("Use --pages to specify file input.");
		return (DataSet<Long>)env.generateSequence(1L, params.getInt("numPages"));
	}

	private static DataSet<Tuple2<Long, Long>> getLinksDataSet(final ExecutionEnvironment env, final ParameterTool params) {
		if (params.has("links")) {
			return (DataSet<Tuple2<Long, Long>>)env.readCsvFile(params.get("links")).fieldDelimiter(",").lineDelimiter("\n").types(Long.class, Long.class);
		}
		System.out.println("Executing PageRank example with default links data set.");
		System.out.println("Use --links to specify file input.");
		return PageRankData.getDefaultEdgeDataSet(env);
	}

	private static DataSet<Tuple2<Long, Double>> getCompensationdata(final ExecutionEnvironment env, final ParameterTool params) {
		final Configuration parameters = new Configuration();
		parameters.setBoolean("recursive.file.enumeration", true);
		final String readpath = params.get("compensationDataPath");
		return env.readTextFile(readpath).withParameters(parameters).map(new MapFunction<String, Tuple2<Long, Double>>() {
			@Override
			public Tuple2<Long, Double> map(final String s) {
				final String[] a = s.split(",");
				final Long x = Long.parseLong(a[0].substring(1, a[0].length()));
				final Double y = Double.parseDouble(a[1].substring(0, a[1].length() - 1));
				return new Tuple2<Long, Double>(x, y);
			}
		});
	}

	private static DataSet<Tuple2<Long, Long>> getLinksDataSet_txt(final ExecutionEnvironment env, final ParameterTool params) {
		final Configuration parameters = new Configuration();
		parameters.setBoolean("recursive.file.enumeration", true);
		if (params.has("links")) {
			return env.readTextFile(params.get("links")).withParameters(parameters).map(new MapFunction<String, Tuple2<Long, Long>>() {
				@Override
				public Tuple2<Long, Long> map(final String value) throws Exception {
					final String[] a = value.split("\\s+");
					return new Tuple2<Long, Long>(Long.parseLong(a[0]), Long.parseLong(a[1]));
				}
			});
		}
		System.out.println("Executing PageRank example with default links data set.");
		System.out.println("Use --links to specify file input.");
		return PageRankData.getDefaultEdgeDataSet(env);
	}

	public static final class addToPages implements JoinFunction<Tuple1<Long>, Tuple2<Long, Long[]>, Tuple2<Long, Long[]>>
	{
		Long[] topages;

		public addToPages(final Long[] pages) {
			this.topages = pages;
		}

		@Override
		public Tuple2<Long, Long[]> join(final Tuple1<Long> x, final Tuple2<Long, Long[]> y) {
			if (y != null) {
				return y;
			}
			return new Tuple2<Long, Long[]>(x.f0, this.topages);
		}
	}

	public static final class RankAssigner implements MapFunction<Long, Tuple2<Long, Double>>, Serializable
	{
		Tuple2<Long, Double> outPageWithRank;

		public RankAssigner(final double rank) {
			this.outPageWithRank = new Tuple2<Long, Double>(-1L, rank);
		}

		@Override
		public Tuple2<Long, Double> map(final Long page) {
			this.outPageWithRank.f0 = page;
			return this.outPageWithRank;
		}
	}

	@FunctionAnnotation.ForwardedFields({ "0" })
	public static final class BuildOutgoingEdgeList implements GroupReduceFunction<Tuple2<Long, Long>, Tuple2<Long, Long[]>>, Serializable
	{
		private final ArrayList<Long> neighbors;

		public BuildOutgoingEdgeList() {
			this.neighbors = new ArrayList<Long>();
		}

		@Override
		public void reduce(final Iterable<Tuple2<Long, Long>> values, final Collector<Tuple2<Long, Long[]>> out) {
			this.neighbors.clear();
			Long id = 0L;
			for (final Tuple2<Long, Long> n : values) {
				id = n.f0;
				this.neighbors.add(n.f1);
			}
			out.collect(new Tuple2<Long, Long[]>(id, this.neighbors.toArray(new Long[this.neighbors.size()])));
		}
	}

	public static final class JoinVertexWithEdgesMatch implements FlatMapFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>>, Tuple2<Long, Double>>
	{
		private long numPages;

		public JoinVertexWithEdgesMatch(final long num) {
			this.numPages = 1L;
			this.numPages = num;
		}

		@Override
		public void flatMap(final Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>> value, final Collector<Tuple2<Long, Double>> out) {
			if (value.f1 != null) {
				final Long[] neighbors = value.f1.f1;
				final double rank = value.f0.f1;
				final double rankToDistribute = rank / neighbors.length;
				for (final Long neighbor : neighbors) {
					out.collect(new Tuple2<Long, Double>(neighbor, rankToDistribute));
				}
				out.collect(new Tuple2<Long, Double>(value.f0.f0, 0.0));
			}
			else {
				out.collect(new Tuple2<Long, Double>(-1L, value.f0.f1 / this.numPages));
			}
		}
	}

	@FunctionAnnotation.ForwardedFields({ "0" })
	public static final class Dampener implements MapFunction<Tuple2<Long, Double>, Tuple2<Long, Double>>, Serializable
	{
		private final double dampening;
		private final double randomJump;

		public Dampener(final double dampening, final double numVertices) {
			this.dampening = dampening;
			this.randomJump = (1.0 - dampening) / numVertices;
		}

		@Override
		public Tuple2<Long, Double> map(final Tuple2<Long, Double> value) {
			value.f1 = value.f1 * this.dampening + this.randomJump;
			return value;
		}
	}

	public static final class crossPages implements CrossFunction<Tuple2<Long, Double>, Tuple2<Long, Double>, Tuple2<Long, Double>>
	{
		private final double dampening;
		private final double randomJump;

		public crossPages(final double dampening, final double numVertices) {
			this.dampening = dampening;
			this.randomJump = (1.0 - dampening) / numVertices;
		}

		@Override
		public Tuple2<Long, Double> cross(final Tuple2<Long, Double> x, final Tuple2<Long, Double> y) {
			x.f1 = this.dampening * (x.f1 + y.f1) + this.randomJump;
			return x;
		}
	}

	public static final class EpsilonFilter implements FilterFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>>>
	{
		private double EPSILON;

		public EpsilonFilter(final double eps) {
			this.EPSILON = eps;
		}

		@Override
		public boolean filter(final Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) {
			return !value.f0.f0.equals(-1L) && Math.abs(value.f0.f1 - value.f1.f1) > this.EPSILON;
		}
	}

	private static class PointAssigner implements JoinFunction<Tuple2<Long, Double>, Tuple2<Long, Double>, Tuple2<Long, Double>> {
		double r_nonfailed;
		long n_nonfailed;
		int numPages;

		public PointAssigner(final double r_nonfailed, final Long n_nonfailed, final Integer numPages) {
			this.r_nonfailed = r_nonfailed;
			this.n_nonfailed = n_nonfailed;
			this.numPages = numPages;
		}

		@Override
		public Tuple2<Long, Double> join(final Tuple2<Long, Double> pageWithRanks, final Tuple2<Long, Double> compensationData) {
			if (compensationData != null) {
				return new Tuple2<Long, Double>(pageWithRanks.f0, (this.n_nonfailed * compensationData.f1) / (this.numPages * this.r_nonfailed));
			} else {
				return new Tuple2<Long, Double>(pageWithRanks.f0, 1.0d / this.numPages);
			}
		}
	}

	public static final class toTuple implements MapFunction<Long, Tuple1<Long>> {
		@Override
		public Tuple1<Long> map(final Long x) {
			return new Tuple1<Long>(x);
		}
	}
}